package ksv.cron.v1;

import ksv.cron.CronConfig;
import ksv.cron.CronContext;
import ksv.cron.CronModel;
import ksv.cron.bean.SysCron;
import lombok.Getter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;
import reactor.core.publisher.Flux;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @创建人 wsdc
 * @时间 2021/1/31
 * @描述
 */
@Getter
public class Looper implements InitializingBean, DisposableBean, ApplicationContextAware {
    @Resource
    CronContainer cronContainer;

    @Resource
    CronConfig cronConfig;

    @Resource
    Locker locker;

    Map<String, CronCallback> cbMap = new HashMap<>();

    List<CronCallbackPostProcessor> cronCallbackPostProcessorList = new LinkedList<>();

    List<CronLoader> cronLoaderList = new LinkedList<>();

    List<CronMonitor> cronMonitorList = new LinkedList<>();

    List<CronLoadPostProcessor> cronLoadPostProcessorList = new LinkedList<>();

    transient boolean working = true;

    ExecutorService executor;

    LinkedBlockingQueue<Runnable> workQueue;

    ApplicationContext applicationContext;

    public void start() {
        int size = cronConfig.getThreadCount();
        workQueue = new LinkedBlockingQueue<>();
        executor = new ThreadPoolExecutor(size, size, 30, TimeUnit.SECONDS, workQueue, new ThreadFactory() {
            private AtomicInteger ai = new AtomicInteger(0);

            @Override
            public Thread newThread(java.lang.Runnable r) {
                Thread t = new Thread(r);
                t.setName("cron-looper-" + ai.incrementAndGet());
                return t;
            }
        });

        onCronLoaded();
        onReportContainerInfo();
        onReportTaskQueueInfo();

        loop();

    }

    private void onCronLoaded(){
        //  定时扫描数据
        Flux.interval(Duration.ofSeconds(1), Duration.ofSeconds(cronConfig.getLoadCronFrequency()))
                .subscribe(k -> {
                    //  异步
                    executor.execute(() -> {
                        loadCron();
                    });
                });
    }

    private void onReportContainerInfo(){
        //  定时报备下容器里面的信息
        Flux.interval( Duration.ofSeconds(cronConfig.getLoadCronFrequency()))
                .subscribe(k -> {
                    for (CronMonitor cronMonitor : cronMonitorList) {
                        cronMonitor.containerInfo(cronContainer);
                    }
                });
    }

    private void onReportTaskQueueInfo(){
        //  定时报备阻塞队列的信息
        Flux.interval(Duration.ofSeconds(cronConfig.getLoadCronFrequency()))
                .subscribe(k -> {
                    //  异步
                    executor.execute(() -> {
                        for (CronMonitor cronMonitor : cronMonitorList) {
                            cronMonitor.blockingQueueInfo(workQueue);
                        }
                    });
                });
    }

    private void loop(){
        //  根据cronContainer中的数据进行反馈
        //  使用子线程来处理
        DelayQueue<ICron> queue = cronContainer.queue();
        new Thread(() -> {
            while (working) {
                Exception e0 = null;
                CronContext cronContext = null;
                try {
                    ICron take = queue.take();
                    cronContext = take.currentContext();

                    //  如果需要检查
                    if(take.shouldCheck()){
                        //  如果容器中没有对应的数据 那么 continue
                        if(cronContainer.get(take.sysCron().getId()) == null){
                            continue;
                        }

                        //  如果 cron 处于remove的状态 那么可以 continue
                        if(take.sysCron().isRemove()){
                            cronContainer.remove(take.sysCron().getId());
                            continue;
                        }
                    }

                    //  异步回调
                    final CronContext cronContext1 = cronContext;
                    executor.execute(() -> {
                        onEvent(cronContext1);
                    });

                    //   如果可以有下一个 则继续
                    CronContext next = take.next();

                    if (next != null) {
                        //  如果还有下一个
                        queue.offer(take);
                    } else {
                        //  发布结束
                        executor.execute(() -> {
                            for (CronMonitor v : cronMonitorList) {
                                v.complete(cronContext1);
                            }
                        });

                        //  容器中移除
                        cronContainer.remove(take.sysCron().getId());
                        continue;
                    }

                } catch (Exception e) {
                    e0 = e;
                }

                //   异常发布处理
                if (e0 != null) {
                    for (CronMonitor cronMonitor : cronMonitorList) {
                        cronMonitor.warn(cronContext, e0);
                    }
                }
            }
        })
                .start();
    }

    //  发布一个需要执行的定时任务
    private void onEvent(CronContext context) {
        Exception e0 = null;
        boolean hasLock = locker.lock(context);
        if (!hasLock) {
            context.setHasLock(true);
            return;
        }

        try {
            //  回调前的钩子
            for (CronCallback value : cbMap.values()) {
                for (CronCallbackPostProcessor v : cronCallbackPostProcessorList) {
                    final CronCallbackPostProcessor v1 = v;

                    //  不能因为钩子影响正常的逻辑
                    exception(context, () -> {
                        v1.beforeCall(context, value);
                    });
                }

                //  执行完前置钩子之后 判断是否需要继续
                //  如果标记未不需要继续就可以撤了
                if (!context.isShouldContinue()) {

                    return;
                }

                //  开始时间
                context.setStartTime(LocalDateTime.now());
                //  执行(包含重试)
                for (int i = 0; i < cronConfig.getTryAgain() + 1; i++) {
                    exception(context, () -> {
                        if (context.getResult() == null) {
                            CronModel.R execute = value.execute0(context);
                            context.setResult(execute);
                        }
                    });
                    //  结束时间
                    context.setEndTime(LocalDateTime.now());
                }

                //  回调后置钩子
                for (CronCallbackPostProcessor v : cronCallbackPostProcessorList) {
                    exception(context, () -> {
                        //  根据result 判断成功还是失败
                        v.afterCall(context, value, context.getResult());
                    });
                }
            }
        } catch (Exception e) {
            e0 = e;
        } finally {
            locker.release(context);
        }
    }

    private void exception(CronContext context, Runnable1 r) {
        Exception e0 = null;
        try {
            r.run();
        } catch (Exception e) {
            e0 = e;
        }

        if (e0 != null) {
            for (CronMonitor cronMonitor : cronMonitorList) {
                cronMonitor.warn(context, e0);
            }
        }
    }

    private void loadCron() {
        for (CronLoader cronLoader : cronLoaderList) {
            List<SysCron> provide = cronLoader.provide(cronConfig, LocalDateTime.now());
            for (SysCron sysCron : provide) {
                cronContainer.add(sysCron);
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        working = false;
        executor.shutdown();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //  开始工作
        start();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        //  收集数据
        this.applicationContext = applicationContext;

        //  收集回调
        Map<String, CronCallback> callbackList = applicationContext.getBeansOfType(CronCallback.class);
        for (CronCallback value : callbackList.values()) {
            cbMap.put(value.type(), value);
        }

        //  收集回调的钩子函数
        Map<String, CronCallbackPostProcessor> callbackPostProcessorMap = applicationContext.getBeansOfType(CronCallbackPostProcessor.class);
        for (CronCallbackPostProcessor value : callbackPostProcessorMap.values()) {
            cronCallbackPostProcessorList.add(value);
        }
        AnnotationAwareOrderComparator.sort(cronCallbackPostProcessorList);

        //  收集 CronLoader
        Map<String, CronLoader> cronLoaderMap = applicationContext.getBeansOfType(CronLoader.class);
        for (CronLoader value : cronLoaderMap.values()) {
            cronLoaderList.add(value);
        }
        AnnotationAwareOrderComparator.sort(cronLoaderList);

        //  收集 CronLoadPostProcessor
        Map<String, CronLoadPostProcessor> loadPostProcessorMap = applicationContext.getBeansOfType(CronLoadPostProcessor.class);
        for (CronLoadPostProcessor value : loadPostProcessorMap.values()) {
            cronLoadPostProcessorList.add(value);
        }
        AnnotationAwareOrderComparator.sort(cronLoadPostProcessorList);

        //  收集 Monitor
        Map<String, CronMonitor> cronMonitorMap = applicationContext.getBeansOfType(CronMonitor.class);
        for (CronMonitor value : cronMonitorMap.values()) {
            cronMonitorList.add(value);
        }
        AnnotationAwareOrderComparator.sort(cronMonitorList);
    }

    private static interface Runnable1 {
        void run() throws Exception;
    }
}
